package com.shujia;


import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.lazy.LazyString;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.io.Text;

import java.util.ArrayList;
import java.util.List;

public class MyMovieUDTF extends GenericUDTF {
    /*
{"movie": [{"movie_name": "肖申克的救赎", "MovieType": "犯罪" }, {"movie_name": "肖申克的救赎", "MovieType": "剧情" }]}
{"movie": [{"movie_name": "霸王别姬", "MovieType": "剧情" }, {"movie_name": "霸王别姬", "MovieType": "爱情" } ] }
{"movie": [{"movie_name": "阿甘正传", "MovieType": "剧情" },{"movie_name": "阿甘正传", "MovieType": "爱情" }] }
{"movie": [{"movie_name": "泰坦尼克号", "MovieType": "剧情" },{"movie_name": "泰坦尼克号", "MovieType": "爱情" },{"movie_name": "泰坦尼克号", "MovieType": "灾难" }] }
{"movie": [{"movie_name": "这个杀手不太冷", "MovieType": "剧情" },{"movie_name": "这个杀手不太冷", "MovieType": "动作" },{"movie_name": "这个杀手不太冷", "MovieType": "犯罪" }] }

        create table movie_json (
            movie String
        );

        load data local inpath '/usr/local/soft/hive-3.1.2/jars/movie.txt' into table movie_json;

        select movie_json(movie)  from bigdata.movie_json =>
        肖申克的救赎 犯罪
        肖申克的救赎 剧情
        霸王别姬 剧情
        霸王别姬 爱情


        create temporary function movie_json as 'com.shujia.MyMovieUDTF';


     */

    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
        System.out.println("initialize function running... mysqlCon");

        List<? extends StructField> allStructFieldRefs = argOIs.getAllStructFieldRefs();
        if (allStructFieldRefs.size() != 1) {
            throw new UDFArgumentLengthException("要求传入一个JSON参数:{\"movie\": [{\"movie_name\": \"这个杀手不太冷\", \"MovieType\": \"剧情\" }...");
        }

        // 一个 StructField 可以代表是一个字段的类型
        for (StructField allStructFieldRef : allStructFieldRefs) {
            ObjectInspector fieldObjectInspector = allStructFieldRef.getFieldObjectInspector();

            if (!fieldObjectInspector.getCategory().equals(ObjectInspector.Category.PRIMITIVE)) {
                throw new UDFArgumentTypeException(allStructFieldRef.getFieldID(),"该函数要求传入字符串类型");
            }
        }


        // StructObjectInspector

        ArrayList<String> fieldNames = new ArrayList<String>();
        ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();
        fieldNames.add("movie_name");
        fieldNames.add("movie_type");
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);

    }



    @Override
    public void process(Object[] args) throws HiveException {

        String oneJson = ((LazyString) args[0]).toString();
        // 需要将其转成一个JSON对象
        System.out.println(oneJson);

//        JSONObject jsonObject = new JSONObject().putObject(oneJson);
        JSONObject jsonObject = JSONObject.parseObject(oneJson);
        JSONArray movie = jsonObject.getJSONArray("movie");
        String[] oneLine = new String[2];
        for (int i = 0; i < movie.size(); i++) {
            JSONObject nextJsonObject = movie.getJSONObject(0);
            String movieName = nextJsonObject.getString("movie_name");
            String movieType = nextJsonObject.getString("MovieType");
            oneLine[0] = movieName;
            oneLine[1] = movieType;
            // 将一行数据写出去
            forward(oneLine);
            oneLine[0] = null;
            oneLine[1] = null;
        }
    }

    @Override
    public void close() throws HiveException {
//        System.out.println("No thing to do..");
    }
}
